package com.shujia.opt

import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SparkSession
import org.apache.spark.{Partitioner, SparkContext}

object Code02SparkDoubleAggData {
  def main(args: Array[String]): Unit = {
    /**
     * Spark数据倾斜：
     * 表现：在Job任务执行时，部分的Task一直处于运行状态，而其他的Task已经处理完成
     *
     * 方式2：双重聚合操作
     *
     */


    val spark: SparkSession = SparkSession
      .builder()
      .master("local")
      .appName("spark")
      .getOrCreate()

    val sc: SparkContext = spark.sparkContext
    val data: RDD[String] = sc.textFile("spark_code/data/word/*")

    data
      .flatMap(_.split(","))
      .map {
        case word => {
//          println(s"=>${(math.floor(math.random()*10)%5)+1}_$word")
          (s"${(math.floor(math.random()*10)%5)+1}_$word",1)
        }
      }
      .reduceByKey(_+_,5) // 第一次聚合
      .map{
        case (key,value) => {
//          println(s"+>${key.substring(4)}")
          (key.substring(4),value)
        }
      }
      .reduceByKey(_+_,2) // 第二次聚合
      .foreach(println)


    // (NULL,23680)
    while (true) {}
  }
}
